package com.xzx.spark.core.transform

import org.apache.spark.rdd.RDD
import org.apache.spark.{SparkConf, SparkContext}

/**
 *
 *
 *
 * @author xinzhixuan
 * @version 1.0
 * @date 2021-06-26 8:39 下午
 */
object Spark020_KeyValue_CombineByKey {
  def main(args: Array[String]): Unit = {
    val conf = new SparkConf().setMaster("local[*]").setAppName("Spark020_KeyValue_CombineByKey")
    val context = new SparkContext(conf)
    //将数据 List(("a", 88), ("b", 95), ("a", 91), ("b", 93), ("a", 95), ("b", 98))求每个 key 的平均值
    val rdd: RDD[(String, Int)] = context.makeRDD(List(("a", 88), ("b", 95), ("a", 91), ("b", 93), ("a", 95), ("b", 98)), 2)
    val combineByKeyRdd: RDD[(String, (Int, Int))] = rdd.combineByKey(
      (_, 1), // 对分区内第一个元素进行结构转换
      (acc: (Int, Int), v: Int) => (acc._1 + v, acc._2 + 1),
      (acc1: (Int, Int), acc2: (Int, Int)) => (acc1._1 + acc2._1, acc1._2 + acc2._2)
    )
    combineByKeyRdd.map(x => (x._1, x._2._1 / x._2._2)).collect().foreach(println)

    context.stop()
  }
}
